package com.yc.bigdata.flink.demo.util;

import com.alibaba.fastjson.JSON;
import com.yc.bigdata.flink.demo.vo.UserAction;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-02-11 11:51
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-02-11 11:51    YuanChilde     1.0        新增
 */
public class KafkaSendDemoUtil {
    public static final String broker_list = "dev.dongbaosoft.com:9092";
    public static final String topic = "bigdata-test-in";

    public static void writeToKafka() throws InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", broker_list);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer producer = new KafkaProducer<String, String>(props);

        for (int i = 1; i <= 100; i++) {
            UserAction userAction = new UserAction(i+"", 1293984000, "click", "productID1", 10);
            ProducerRecord record = new ProducerRecord(topic, JSON.toJSONString(userAction));
            producer.send(record);
            Thread.sleep(10 * 1000); //发送一条数据 sleep 10s，相当于 1 分钟 6 条
        }
        producer.flush();
    }

    public static void main(String[] args) throws InterruptedException {
        writeToKafka();
    }
}
